package com.lu.sb211201rocketmq.listener;

import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBufUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 *  这里不是用的 rocketmq-spring-boot-starter 框架里的类，用的是 rocketmq-client 基础的类
 *  MessageListenerOrderly：有序消费
 *  MessageListenerConcurrently：并发消费
 */

@Component
@Slf4j
public class MQConsumeMsgListenerProcessor implements MessageListenerOrderly {

    /**
     * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息<br/>
     * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
     */
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        if (CollectionUtils.isEmpty(list)) {
            log.info("message is empty, wait for next message!");
            return ConsumeOrderlyStatus.SUCCESS;
        }
        log.info("message body:" + ByteBufUtil.hexDump(list.get(0).getBody()));
        // TODO 逻辑处理
        MessageExt messageExt = list.get(0);
        System.out.println(messageExt);
        JSONObject messageBody = JSONObject.parseObject(new String(messageExt.getBody(), StandardCharsets.UTF_8));
        System.out.println(messageBody);

        // 如果没有return success ，consumer会重新消费该消息，直到return success
        return ConsumeOrderlyStatus.SUCCESS;
    }

}
